package com.example.flinkcourse.lesson3.event;

import com.example.flinkcourse.lesson3.model.Transaction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 交易事件处理器
 * 使用 KeyedProcessFunction 处理交易事件
 */
public class TransactionEventHandler extends KeyedProcessFunction<String, Transaction, Alert> {
    private static final Logger LOG = LoggerFactory.getLogger(TransactionEventHandler.class);
    
    private ValueState<Double> totalAmountState;
    private ValueState<Long> lastAlertTimeState;
    
    @Override
    public void open(Configuration parameters) {
        LOG.info("Initializing transaction event handler...");
        
        // 初始化状态
        ValueStateDescriptor<Double> amountDescriptor = 
            new ValueStateDescriptor<>("total-amount", Types.DOUBLE);
        totalAmountState = getRuntimeContext().getState(amountDescriptor);
        
        ValueStateDescriptor<Long> timeDescriptor = 
            new ValueStateDescriptor<>("last-alert-time", Types.LONG);
        lastAlertTimeState = getRuntimeContext().getState(timeDescriptor);
    }
    
    @Override
    public void processElement(Transaction transaction, Context ctx, Collector<Alert> out) {
        LOG.debug("Processing transaction: {}", transaction.getId());
        
        try {
            // 获取当前状态
            Double totalAmount = totalAmountState.value();
            if (totalAmount == null) {
                totalAmount = 0.0;
            }
            
            // 更新总金额
            totalAmount += transaction.getAmount();
            totalAmountState.update(totalAmount);
            
            // 检查是否需要告警
            Long lastAlertTime = lastAlertTimeState.value();
            if (lastAlertTime == null) {
                lastAlertTime = 0L;
            }
            
            // 如果总金额超过阈值且距离上次告警超过1小时
            if (totalAmount > 10000.0 && 
                (ctx.timestamp() - lastAlertTime) > Time.hours(1).toMilliseconds()) {
                
                // 生成告警
                Alert alert = Alert.builder()
                    .accountId(transaction.getAccountId())
                    .message("High amount alert: " + totalAmount)
                    .timestamp(ctx.timestamp())
                    .build();
                
                out.collect(alert);
                
                // 更新最后告警时间
                lastAlertTimeState.update(ctx.timestamp());
                
                LOG.info("Alert generated for account {}: {}", 
                    transaction.getAccountId(), alert.getMessage());
            }
        } catch (Exception e) {
            LOG.error("Error processing transaction: " + transaction.getId(), e);
        }
    }
}

/**
 * 告警数据模型
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
class Alert {
    private String accountId;
    private String message;
    private long timestamp;
} 